package com.study.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/25 20:53
 */
public class UdfAggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        //1.在创建表的DDL中直接定义时间属性：事件时间
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " event_time AS TO_TIMESTAMP( FROM_UNIXTIME( ts / 1000 ) ), " +   //将ts时间戳转换成timestamp类型
                " WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND " +  //设置水位线的时间间隔为1s
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";
        tabEnv.executeSql(createDDL);

        //2.注册自定义聚合函数
        tabEnv.createTemporarySystemFunction("WeightAvgFunction",WeightAvgFunction.class);

        //3.调用UDF进行查询转换
        Table table = tabEnv.sqlQuery("select user,WeightAvgFunction(ts,1) as w_avg from click_table group by user");

        //4.转换成流打印输出
        tabEnv.toChangelogStream(table).print("result：");

        env.execute();
    }

    public static class WeightAvgAccumulator{
        public long sum = 0L;
        public int count = 0;
    }

    //实现自定义的聚合函数，计算加权平均值
    public static class  WeightAvgFunction extends AggregateFunction<Long,WeightAvgAccumulator>{
        @Override
        public WeightAvgAccumulator createAccumulator() {
            //定义累加器
            return new WeightAvgAccumulator();
        }

        @Override
        public Long getValue(WeightAvgAccumulator accumulator) {
            //获取结果
            if (accumulator.count == 0){
                return null;
            }else {
                return accumulator.sum / accumulator.count;
            }
        }

        //累加计算的方法
        public void accumulate(WeightAvgAccumulator accumulator,Long iValue,Integer iWeight){
            accumulator.sum += iValue * iWeight;
            accumulator.count += iWeight;
        }

    }
}
